package com.wyl.learn03;

import cn.hutool.core.date.StopWatch;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wyl.config.RabbitMqConfig;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认，多条消息异步发布确认
 */
public class Producer03_3 {
    public static final String TASK_QUEUE_NAME_CONFORM = "ack_queue_confirm";

    public static void main(String[] argv) throws Exception {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMqConfig.IP);
        factory.setUsername(RabbitMqConfig.USERNAME);
        factory.setPassword(RabbitMqConfig.PASSWORD);

        // 要发送的消息数量
        int MESSAGE_COUNT = 102;

        //channel 实现了自动 close 接口 自动关闭 不需要显示关闭
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明非持久化队列
        channel.queueDeclare(TASK_QUEUE_NAME_CONFORM, false, false, false, null);
        // 开启发布确认
        channel.confirmSelect();
        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        /**
         * 确认收到消息的一个回调
         * 1.消息序列号
         * 2.是否批量确认。true 可以确认小于等于当前序列号的消息；false 确认当前序列号消息。在RabbitMQ中，Confirm模式的批量确认是由服务器自动进行的，无法通过配置或参数设置来控制。
         */
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                // 返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                // 清除该部分未确认消息
                confirmed.clear();
            } else {
                // 只清除当前序列号的消息
                outstandingConfirms.remove(deliveryTag);
            }
        };
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("发布的消息" + message + "未被确认，序列号" + deliveryTag);
        };
        /**
         * 添加一个异步确认的监听器
         * 1.确认收到消息的回调
         * 2.未收到消息的回调
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            /**
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             */
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", TASK_QUEUE_NAME_CONFORM, null, message.getBytes());
        }
        System.out.println("发布" + MESSAGE_COUNT + "个批量确认消息,耗时" + stopWatch.getTotalTimeMillis() + "ms");
    }
}